package com.adcc.ags.cgw.proxy;

import com.adcc.ags.cgw.configuration.GWFactory;
import com.adcc.ags.cgw.configuration.UserListFactory;
import com.adcc.ags.cgw.entity.mq.MQInfo;
import com.adcc.ags.cgw.entity.mq.MQState;
import com.adcc.ags.cgw.entity.msg.TrafficMsg;
import com.adcc.ags.cgw.entity.user.*;
import com.adcc.utility.log.Log;
import com.adcc.utility.mq.configuration.MQConfiguration;
import com.adcc.utility.mq.configuration.MQConfigurationFactory;
import com.adcc.utility.mq.entity.Message;
import com.adcc.utility.mq.transfer.*;
import com.adcc.utility.mq.transfer.ibm.IBMMQConnectionPool;
import com.adcc.utility.mq.transfer.ibm.IBMMQTransfer;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import io.dropwizard.setup.Environment;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.joda.time.Seconds;
import org.joda.time.format.DateTimeFormat;

import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * 网关用户代理
 */
public class UserProxy extends BaseProxy implements TopicMsgListener, MQStateListener {

    // 网关心跳发送线程
    private static final String CONSTANT_GW_HB_SEND_THREAD_NAME = "Thread.UserProxy.GWHbSend";

    // 用户心跳检测线程
    private static final String CONSTANT_USER_HB_CHECK_THREAD_NAME = "Thread.UserProxy.UserHbCheck";

    // 单例对象
    private static UserProxy instance;

    // 网关模式
    private GWMode gwMode = GWMode.NORMAL;

    // MQ连接状态
    private MQInfo mqInfo = new MQInfo();

    // 网关配置工厂
    private GWFactory gwFactory;

    // MQConfiguration
    private MQConfiguration configuration;

    // MQConnectionPool
    private MQConnectionPool connectionPool;

    // MQTransfer
    private MQTransfer transfer;

    // 用户信息Map
    private Map<String, UserInfo> userInfoMap = Maps.newConcurrentMap();

    // 心跳发布Runnable
    private HbSendRunnable hbSendRunnable = new HbSendRunnable();

    // 心跳检测Runnable
    private HbCheckRunnable hbCheckRunnable = new HbCheckRunnable();

    /**
     * 构造函数
     */
    private UserProxy() {

    }

    /**
     * 单例方法
     * @return
     */
    public synchronized static UserProxy getInstance() {
        if (instance == null) {
            instance = new UserProxy();
        }
        return instance;
    }

    public GWMode getGWMode(){
        return gwMode;
    }

    public void setGWMode(GWMode gwMode) {
        this.gwMode = gwMode;
    }

    public MQInfo getMqInfo() {
        return mqInfo;
    }

    public Map<String, UserInfo> getUserInfoMap() {
        return userInfoMap;
    }

    /**
     * 建立Managed
     * @param environment
     */
    public void build(Environment environment, GWFactory gwFactory) throws Exception{
        if (environment == null) {
            throw new NullPointerException("environment is null");
        }
        if (gwFactory == null) {
            throw new NullPointerException("gwFactory is null");
        }
        this.gwFactory = gwFactory;
        environment.lifecycle().manage(this);
    }

    /**
     * 处理用户心跳信息
     * @param hbInfo
     */
    private void processUserHb(String hbInfo) {
        try {

            // 格式化心跳消息
            UserInfo userInfo = new UserInfo();
            userInfo.formatJson(hbInfo);

            // 检查发送时间是否有效
            if(Strings.isNullOrEmpty(userInfo.getSendTime())){
                return;
            }

            // 检查用户ID、用户名称、用户IP是否存在
            if(Strings.isNullOrEmpty(userInfo.getId()) || Strings.isNullOrEmpty(userInfo.getName()) || Strings.isNullOrEmpty(userInfo.getIp())){
                return;
            }

            // 检查用户是否包含在用户列表内
            if(userInfoMap.containsKey(userInfo.getId())){

                // 检查用户名称是否一致
                if(!userInfoMap.get(userInfo.getId()).getName().equals(userInfo.getName())){
                    return;
                }

                // 验证发送时间是否有效
                if(Strings.isNullOrEmpty(userInfo.getSendTime())){
                    return;
                }

                // 更新用户状态
                if(userInfoMap.get(userInfo.getId()).getState() != UserState.ONLINE){
                    userInfoMap.get(userInfo.getId()).setState(UserState.ONLINE);
                    userInfoMap.get(userInfo.getId()).setUpdateTime(DateTime.now(DateTimeZone.UTC).toString(DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss")));
                }

                // 更新用户IP
                userInfoMap.get(userInfo.getId()).setIp(userInfo.getIp());

                // 更新用户心跳发送时间
                userInfoMap.get(userInfo.getId()).setSendTime(DateTime.now(DateTimeZone.UTC).toString(DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss")));
            }
        } catch (Exception ex) {
            Log.error(UserProxy.class.getName(), "processUserHb() error", ex);
        }
    }

    /**
     * 发送网关心跳信息
     * @param hbInfo
     */
    private void sendGWHb(String hbInfo) {
        try {
            Message message = new Message(hbInfo.getBytes());
            ((IBMMQTransfer) transfer).sendTopic(gwFactory.getHeartbeatFactory().getHbSendTopic(), message, (int) gwFactory.getHeartbeatFactory().getHbSendTimeout());
            Log.info(UserProxy.class.getName(), "send gw heartbeat:\r\n" + hbInfo);
        } catch (Exception ex) {
            Log.error(UserProxy.class.getName(), "sendGWHb() error", ex);
        }
    }

    /**
     * 读取用户列表
     */
    public void loadUserList() {
        try{
            if(UserListFactory.getInstance().getUserList() == null || UserListFactory.getInstance().getUserList().size() == 0){
                userInfoMap.clear();
            }else{
                List<String> list = Lists.newArrayListWithExpectedSize(userInfoMap.size());
                for(Iterator<String> iterator = userInfoMap.keySet().iterator();iterator.hasNext();){
                    boolean flag = false;
                    String strUserId = iterator.next();
                    for(User user : UserListFactory.getInstance().getUserList()){
                        if(strUserId.equals(user.getId())){
                            flag = true;
                            break;
                        }
                    }
                    if(!flag){
                        list.add(strUserId);
                    }
                }
                for(User user : UserListFactory.getInstance().getUserList()){
                    if(!userInfoMap.containsKey(user.getId())){
                        UserInfo userInfo = new UserInfo(user.getId(), user.getName());
                        userInfoMap.put(user.getId(), userInfo);
                    }
                }
                if(list.size() > 0){
                    for(String strUserId : list){
                        userInfoMap.remove(strUserId);
                    }
                }
            }
        }catch (Exception ex){
            Log.error(UserListFactory.class.getName(),"loadUserList() error",ex);
        }
    }

    /**
     * 添加用户
     * @param user
     */
    public void addUser(User user){
        if(user != null){
            if(!userInfoMap.containsKey(user.getId())){
                UserInfo userInfo = new UserInfo(user.getId(), user.getName());
                userInfoMap.put(user.getId(), userInfo);
            }
        }
    }

    /**
     * 删除用户
     * @param userId
     */
    public void removeUser(String userId){
        if(!Strings.isNullOrEmpty(userId)){
            if(userInfoMap.containsKey(userId)){
                userInfoMap.remove(userId);
            }
        }
    }

    /**
     * 保存报文
     */
    public void saveMsg(TrafficMsg msg){
        try {
            if(msg.getRawMsg() != null && msg.getRawMsg().length > 0){
                Map<String,Object> map = Maps.newHashMapWithExpectedSize(4);
                map.put("msgId",msg.getId());
                map.put("type",msg.getType());
                map.put("sender", msg.getSender());
                map.put("receiver",msg.getReceiver());
                map.put("flag","0");
                Message message = new Message(map,msg.getRawMsg());
                transfer.sendQueue(gwFactory.getDbForward(), message);
            }else{
                Map<String,Object> map = Maps.newHashMapWithExpectedSize(1);
                map.put("msgId",msg.getId());
                map.put("flag","1");
                String strErrorInfo = msg.toJSON();
                Message message = new Message(map,strErrorInfo.getBytes());
                transfer.sendQueue(gwFactory.getDbForward(), message);
            }
        } catch (Exception ex) {
            Log.error(UserProxy.class.getName(), "saveMsg() error", ex);
        }
    }

    /**
     * 重置报文计数
     */
    public void resetMsgCount(){
        try{
            if(userInfoMap.size() > 0){
                for(Iterator<String> iterator = userInfoMap.keySet().iterator();iterator.hasNext();){
                    String strKey = iterator.next();
                    userInfoMap.get(strKey).setUplink(0l);
                    userInfoMap.get(strKey).setDownlink(0l);
                }
            }
        }catch (Exception ex){
            Log.error(UserProxy.class.getName(),"resetMsgCount() error",ex);
        }
    }

    /**
     * 保存报文信息
     */
    public void saveMsgInfo(){
        try{

        }catch (Exception ex){
            Log.error(UserProxy.class.getName(),"saveMsgInfo() error",ex);
        }
    }

    @Override
    public void start() {
        try {

            // 初始化网关用户
            if(userInfoMap.size() > 0){
                userInfoMap.clear();
            }
            loadUserList();

            // 设置MQ连接配置
            configuration = MQConfigurationFactory.getInstance().createIBMMQConfiguration(gwFactory.getMqFactory().getHost(),
                    gwFactory.getMqFactory().getPort(),
                    gwFactory.getMqFactory().getQueueManager(),
                    gwFactory.getMqFactory().getChannel());

            // 设置MQ连接池配置
            if(gwFactory.getMqFactory().getMaxConnections() > 0 && gwFactory.getMqFactory().getMaxIdelConnections() > 0){
                connectionPool = MQConnectionPoolFactory.getInstance().createIBMMQConnectionPool();
                ((IBMMQConnectionPool)connectionPool).setActiveMode(gwFactory.getMqFactory().getActiveMode());
                ((IBMMQConnectionPool)connectionPool).setTimeout(gwFactory.getMqFactory().getTimeout());
                ((IBMMQConnectionPool)connectionPool).setMaxConnections(gwFactory.getMqFactory().getMaxConnections());
                ((IBMMQConnectionPool)connectionPool).setMaxIdelConnections(gwFactory.getMqFactory().getMaxIdelConnections());
                connectionPool.init(configuration);
            }

            // 创建MQTransfer
            transfer = MQTransferFactory.getInstance().createIBMMQTransfer("Transfer.GW",configuration,connectionPool);

            // 设置Topic监听
            transfer.setTopicListener(this, gwFactory.getHeartbeatFactory().getHbRecvTopic());

            // 设置MQ监听
            transfer.setMQStateListener(this);
            transfer.startAsync();

            // 启动网关心跳发送线程
            hbSendRunnable = new HbSendRunnable();
            new Thread(hbSendRunnable, CONSTANT_GW_HB_SEND_THREAD_NAME).start();

            // 启动用户心跳检测线程
            hbCheckRunnable = new HbCheckRunnable();
            new Thread(hbCheckRunnable, CONSTANT_USER_HB_CHECK_THREAD_NAME).start();
        } catch (Exception ex) {
            Log.error(UserProxy.class.getName(), "start() error",ex);
        }
    }

    @Override
    public void stop() {
        try {

            // 销毁MQTransfer
            transfer.stopAsync();
            MQTransferFactory.getInstance().disposeMQTransfer("Transfer.GW");

            // 停止心跳发送线程
            if (hbSendRunnable != null) {
                hbSendRunnable.close();
            }

            // 停止心跳检测线程
            if (hbCheckRunnable != null) {
                hbCheckRunnable.close();
            }
        } catch (Exception ex) {
            Log.error(UserProxy.class.getName(), "stop() error",ex);
        }
    }

    @Override
    public void onTopicMsg(String tName, Message message, Map<String, String> mqInfo) {
        try{
            if (message != null && message.getLength() > 0) {
                Log.info(UserProxy.class.getName(), "receive from topic:" + tName + "\r\nmessage:\r\n" + message.toString());
                processUserHb(new String(message.getContent(), 0, message.getContent().length));
            }
        }catch (Exception ex){
            Log.error(UserProxy.class.getName(), "onMessage() error",ex);
        }
    }

    @Override
    public void onState(com.adcc.utility.mq.entity.MQState state, Map<String, String> mqInfo) {
        try {
            Log.info(UserProxy.class.getName(), "MQ state:" + mqInfo + " state:" + state);
            if (state == com.adcc.utility.mq.entity.MQState.CONNECTED) {
                this.mqInfo.setState(MQState.CONNECTED);
            } else if(state == com.adcc.utility.mq.entity.MQState.CONNECTING) {
                this.mqInfo.setState(MQState.CONNECTING);
            } else {
                this.mqInfo.setState(MQState.DISCONNECTED);
            }
        }catch (Exception ex){
            Log.error(UserProxy.class.getName(), "onState() error", ex);
        }
    }

    /**
     * 网关心跳发送线程
     */
    class HbSendRunnable implements Runnable {

        // 线程启动标识
        private boolean isStarted;

        /**
         * 关闭线程
         */
        public void close() {
            isStarted = false;
        }

        @Override
        public void run() {
            try {
                Log.info(HbSendRunnable.class.getName(), "Thread:" + CONSTANT_GW_HB_SEND_THREAD_NAME + " start...");
                isStarted = true;
                while (isStarted) {
                    try {
                        if (UserListFactory.getInstance().getUserList().size() > 0) {
                            Map<String, Object> map = Maps.newHashMapWithExpectedSize(4);
                            if (AMQSProxy.getInstance().getAMQS().getState() == MQState.CONNECTED && mqInfo.getState() == MQState.CONNECTED) {
                                map.put("state", GWState.AVAILABLE);
                            } else {
                                map.put("state", GWState.UNAVAILABLE);
                            }
                            map.put("userList", UserListFactory.getInstance().getUserList());
                            map.put("sendTime", DateTime.now(DateTimeZone.UTC).toLocalDateTime().toString(DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss")));
                            map.put("mode", gwMode);
                            sendGWHb(new ObjectMapper().writeValueAsString(map));
                        }

                        Thread.sleep(gwFactory.getHeartbeatFactory().getHbSendInterval() * 1000);
                    } catch (Exception ex) {
                        Log.error(HbSendRunnable.class.getName(), "send heartbeat error", ex);
                    }
                }
            } catch (Exception ex) {
                Log.error(HbSendRunnable.class.getName(), "run() error", ex);
            }
        }
    }

    /**
     * 用户心跳检测线程
     */
    class HbCheckRunnable implements Runnable {

        // 线程启动标识
        private boolean isStarted;

        /**
         * 关闭线程
         */
        public void close() {
            isStarted = false;
        }

        @Override
        public void run() {
            try {
                Log.info(HbCheckRunnable.class.getName(), "Thread:" + CONSTANT_USER_HB_CHECK_THREAD_NAME + " start...");
                isStarted = true;
                while (isStarted) {
                    try {
                        for (Map.Entry<String, UserInfo> entry :  userInfoMap.entrySet()) {
                            UserInfo userInfo = entry.getValue();
                            if (userInfo.getState() == UserState.ONLINE) {
                                DateTime curTime = DateTime.now(DateTimeZone.UTC).toLocalDateTime().toDateTime();
                                DateTime sendTime = DateTime.parse(userInfo.getSendTime(),DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss"));
                                if (Math.abs(Seconds.secondsBetween(curTime,sendTime).getSeconds()) > gwFactory.getHeartbeatFactory().getHbRecvExpiredTime()) {
                                    userInfo.setState(UserState.OFFLINE);
                                    userInfo.setUpdateTime(DateTime.now(DateTimeZone.UTC).toString(DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss")));
                                }
                            }
                        }
                        Thread.sleep(2000);
                    } catch (Exception ex) {
                        Log.error(HbCheckRunnable.class.getName(), "check user heartbeat error", ex);
                    }
                }
            } catch (Exception ex) {
                Log.error(HbCheckRunnable.class.getName(), "run() error", ex);
            }
        }
    }
}
